package com.atguigu.canal.clients;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.realtime.constants.TopicConstant;
import com.atguigu.realtime.utils.KafkaProducerUtil;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * Created by Smexy on 2022/7/19
 *
 *  ①先创建一个客户端对象CanalConnector
 *
 * ②使用客户端对象连接 Canal server端
 *
 * ③订阅表
 *
 * ④解析订阅到的数据
 *
 * ⑤将数据写入kafka
 */
public class OrderInfoClient {

    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {

        /*
            ①先创建一个客户端对象CanalConnector

            SocketAddress address：
                        hostname: canal.properties  的 canal.ip = hadoop103
                        port    : canal.properties  的 canal.port = 11111
            String destination:  canal.properties  的 canal.destinations = example(默认)
                            订阅的mysql实例配置文件 instance.properties 所在的目录名
             String username：   没有
              String password：  没有
                不要和 刚才配置的 canal , canal 混淆了。
         */
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop103", 11111),
                "example", null
                , null
        );


        //②使用客户端对象连接 Canal server端
        canalConnector.connect();

        //③订阅表  格式:库名.表名
        canalConnector.subscribe("220309.order_info");

        //④拉取数据，拉到解析订阅到的数据
        while (true){

            Message message = canalConnector.get(100);

            if (message.getId() == -1){
                //没拉到数据  单押
                System.out.println("当前没有新数据，歇5秒再去");
                //歇会
                Thread.sleep(5000);

                //跳过本次循环，开始下次循环
                continue;
            }


            //打印数据
            //System.out.println(message);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {

                //是rowdata，才可能是 insert | update | delete
                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)){

                    parseData(entry.getStoreValue());
                }

            }


        }

    }

    // 只要 order_info表的insert
    // update操作： 变化前和变化后  maxwell:   {data:{变化后}，old:{更新的字段变化前的数据}}
    // insert操作: 变化后
    private static void parseData(ByteString storeValue) throws InvalidProtocolBufferException {

        //  rowChange 代表一行sql造成的 多行 反序列化后的 数据变化
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

        if (rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            //多行数据变化
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

            // rowData: 1行数据变化
            for (CanalEntry.RowData rowData : rowDatasList) {

                //构建一个JSONObject 希望把每一行封装为一个{}
                JSONObject jsonObject = new JSONObject();

                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

                for (CanalEntry.Column column : afterColumnsList) {

                    jsonObject.put(column.getName(), column.getValue());

                }

                //System.out.println(jsonObject);
                KafkaProducerUtil.sendData(jsonObject.toString(), TopicConstant.ORDER_INFO);

            }

        }



    }
}
